package com.zero.ava.reactor;

import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import reactor.core.Environment;
import reactor.core.Reactor;
import reactor.core.configuration.PropertiesConfigurationReader;
import reactor.core.spec.Reactors;
import reactor.event.Event;
import reactor.event.selector.Selectors;
import reactor.function.Consumer;

public class MainTest {

	public static MainTest instance = new MainTest();

	public static MainTest getInstance() {
		return instance;
	}

	PropertiesConfigurationReader configurationReader = new PropertiesConfigurationReader();
	final Environment environment = new Environment(configurationReader);
	Reactor reactor = Reactors.reactor().env(environment).dispatcher("ringBuffer").get();

	public Reactor getReactor() {
		return reactor;
	}

	public static void main(String[] args) {
		MainTest instance2 = MainTest.getInstance();
		Reactor reactor2 = instance2.getReactor();
		for (int i = 0; i < 10; i++) {
			reactor2.on(Selectors.$("topic" + i), new Zero("I am hreo=" + i));
		}

		ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(10);
		for (int i = 0; i < 10; i++) {
			Player player = new Player();
			Thread thread = new Thread(player, "Player" + (i + 1));
			newFixedThreadPool.execute(thread);
		}
	}
}

class Player extends Thread {
	@Override
	public void run() {
		while (true) {
			try {
				MainTest instance2 = MainTest.getInstance();

				Reactor reactor2 = instance2.getReactor();

				ZeroMessage message = new ZeroMessage("now messsage==" + Thread.currentThread().getId());

				Random random = new Random();

				reactor2.notify("topic" + random.nextInt(10), Event.wrap(message));

			} catch (Exception e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
}

class Zero implements Consumer<ZeroMessage> {

	private String name;

	public Zero(String name) {
		super();
		this.name = name;
	}

	@Override
	public void accept(ZeroMessage t) {
		System.out.println(t.getData());
		try {
			System.out.println(name + "===" + Thread.currentThread().getId() + "==" + Thread.currentThread().getName());
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}

class ZeroMessage extends Event<String> {

	public ZeroMessage(String data) {
		super(data);
	}

	/**
	 * 
	 */
	private static final long serialVersionUID = 5949928024266312608L;
}